package com.aispeech.task.api.service.impl;

import com.aispeech.task.api.data.vo.TaskProgress;
import com.aispeech.task.api.service.TaskManageService;
import com.aispeech.task.common.config.RedisConfig;
import com.aispeech.task.common.data.constant.Progress;
import com.aispeech.task.common.data.domain.Task;
import com.aispeech.task.common.data.enums.APICode;
import com.aispeech.task.common.data.vo.response.Result;
import com.aispeech.task.common.utils.TaskUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import static com.aispeech.task.common.config.RedisConfig.HASH_TASK_KEY;

@Service
@Slf4j
public class TaskManageServiceImpl implements TaskManageService {
    @Autowired
    @Qualifier("hashTaskOperations")
    private HashOperations<String, String, Task> opHash;

    @Autowired
    private StreamOperations<String, String, String> opStream;

    private Task getTaskFromRedis(String taskId, String productId) {
        return opHash.get(HASH_TASK_KEY, TaskUtils.field(taskId, productId));
    }

    @Override
    public Result query() {
        List<Task> tasks = new ArrayList<>();
        opHash.entries(HASH_TASK_KEY).forEach((key, task) -> {
            if (task.getProgress() < Progress.FINISHED)
                tasks.add(task);
        });
        return Result.ok(tasks);
    }

    @Override
    public Result get(String taskId, String productId) {

        Task task = getTaskFromRedis(taskId, productId);

        if (task != null) {
            return Result.ok(TaskProgress.of(task.getProgress()));
        } else {
            return Result.fail(APICode.NOT_FOUND);
        }
    }

    @Override
    public Result reset(String taskId, String productId) {
        Task task = getTaskFromRedis(taskId, productId);

        if (task != null) {
            String taskField = TaskUtils.field(task.getId(), productId);

            task.setProgress(Progress.IDLE);
            opHash.put(HASH_TASK_KEY, taskField, task);

            opStream.add(StreamRecords.string(new HashMap() {
                {
                    put("taskId", taskField);
                }
            }).withStreamKey(RedisConfig.STREAM_NAME));

            return Result.ok(task);
        } else {
            return Result.fail(APICode.NOT_FOUND);
        }
    }

    @Override
    public Result interrupt(String taskId, String productId) {
        Task task = getTaskFromRedis(taskId, productId);

        if (task != null) {
            String taskField = TaskUtils.field(task.getId(), productId);

            task.setProgress(Progress.FINISHED);
            task.updateError(APICode.INTERRUPTED);

            opHash.put(HASH_TASK_KEY, taskField, task);

            return Result.ok(task);
        } else {
            return Result.fail(APICode.NOT_FOUND);
        }
    }
}
